/*
 * Copyright (c) 2009-2012, Salvatore Sanfilippo <antirez at gmail dot com>
 * All rights reserved.
 *
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions are met:
 *
 *   * Redistributions of source code must retain the above copyright notice,
 *     this list of conditions and the following disclaimer.
 *   * Redistributions in binary form must reproduce the above copyright
 *     notice, this list of conditions and the following disclaimer in the
 *     documentation and/or other materials provided with the distribution.
 *   * Neither the name of Redis nor the names of its contributors may be used
 *     to endorse or promote products derived from this software without
 *     specific prior written permission.
 *
 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
 * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
 * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
 * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
 * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
 * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
 * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
 * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
 * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
 * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
 * POSSIBILITY OF SUCH DAMAGE.
 */

#include "server.h"

/* ================================ MULTI/EXEC ============================== */

/* Client state initialization for MULTI/EXEC */
void initClientMultiState(client *c) {
    c->mstate.commands = NULL;
    c->mstate.count = 0;
    c->mstate.cmd_flags = 0;
    c->mstate.cmd_inv_flags = 0;
}

/* Release all the resources associated with MULTI/EXEC state */
void freeClientMultiState(client *c) {
    int j;

    for (j = 0; j < c->mstate.count; j++) {
        int i;
        multiCmd *mc = c->mstate.commands + j;

        for (i = 0; i < mc->argc; i++)
            decrRefCount(mc->argv[i]);
        zfree(mc->argv);
    }
    zfree(c->mstate.commands);
}

/* 将一个新命令添加到事务队列中 */
void queueMultiCommand(client *c) {
    multiCmd *mc;
    int j;
    /* 如果事务已经中止，浪费内存是没有意义的，这在客户端通过管道发送这些消息，
     * 或者懒得阅读之前的响应而没有注意到multi已经中止的情况下很有用。*/
    if (c->flags & CLIENT_DIRTY_EXEC)
        return;
    // 为新数组元素分配空间
    c->mstate.commands = zrealloc(c->mstate.commands, sizeof(multiCmd) * (c->mstate.count + 1));
    // 指向新元素
    mc = c->mstate.commands + c->mstate.count;
    // 设置事务的命令、命令参数数量，以及命令的参数
    mc->cmd = c->cmd;
    mc->argc = c->argc;
    mc->argv = zmalloc(sizeof(robj *) * c->argc);
    memcpy(mc->argv, c->argv, sizeof(robj *) * c->argc);
    for (j = 0; j < c->argc; j++)
        incrRefCount(mc->argv[j]);
    // 事务命令数量计数器增一
    c->mstate.count++;
    c->mstate.cmd_flags |= c->cmd->flags;
    c->mstate.cmd_inv_flags |= ~c->cmd->flags;
}

void discardTransaction(client *c) {
    freeClientMultiState(c);
    initClientMultiState(c);
    c->flags &= ~(CLIENT_MULTI | CLIENT_DIRTY_CAS | CLIENT_DIRTY_EXEC);
    unwatchAllKeys(c);
}

/* Flag the transaction as DIRTY_EXEC so that EXEC will fail.
 * Should be called every time there is an error while queueing a command. */
void flagTransaction(client *c) {
    if (c->flags & CLIENT_MULTI)
        c->flags |= CLIENT_DIRTY_EXEC;
}

void multiCommand(client *c) {
    //禁止嵌套事务
    if (c->flags & CLIENT_MULTI) {
        addReplyError(c, "MULTI calls can not be nested");
        return;
    }
    //打开事务flag
    c->flags |= CLIENT_MULTI;

    addReply(c, shared.ok);
}

void discardCommand(client *c) {
    //说明此时没有事务的情况
    if (!(c->flags & CLIENT_MULTI)) {
        addReplyError(c, "DISCARD without MULTI");
        return;
    }
    //取消事务的逻辑
    discardTransaction(c);
    addReply(c, shared.ok);
}

void beforePropagateMultiOrExec(int multi) {
    if (multi) {
        /* Propagating MULTI */
        serverAssert(!server.propagate_in_transaction);
        server.propagate_in_transaction = 1;
    } else {
        /* Propagating EXEC */
        serverAssert(server.propagate_in_transaction == 1);
        server.propagate_in_transaction = 0;
    }
}

/* Send a MULTI command to all the slaves and AOF file. Check the execCommand
 * implementation for more information. */
void execCommandPropagateMulti(int dbid) {
    beforePropagateMultiOrExec(1);
    propagate(server.multiCommand, dbid, &shared.multi, 1,
              PROPAGATE_AOF | PROPAGATE_REPL);
}

void execCommandPropagateExec(int dbid) {
    beforePropagateMultiOrExec(0);
    propagate(server.execCommand, dbid, &shared.exec, 1,
              PROPAGATE_AOF | PROPAGATE_REPL);
}

/* Aborts a transaction, with a specific error message.
 * The transaction is always aboarted with -EXECABORT so that the client knows
 * the server exited the multi state, but the actual reason for the abort is
 * included too.
 * Note: 'error' may or may not end with \r\n. see addReplyErrorFormat. */
void execCommandAbort(client *c, sds error) {
    discardTransaction(c);

    if (error[0] == '-') error++;
    addReplyErrorFormat(c, "-EXECABORT Transaction discarded because of: %s", error);

    /* Send EXEC to clients waiting data from MONITOR. We did send a MULTI
     * already, and didn't send any of the queued commands, now we'll just send
     * EXEC so it is clear that the transaction is over. */
    if (listLength(server.monitors) && !server.loading)
        replicationFeedMonitors(c, server.monitors, c->db->id, c->argv, c->argc);
}

void execCommand(client *c) {
    int j;
    robj **orig_argv;
    int orig_argc;
    struct redisCommand *orig_cmd;
    int was_master = server.masterhost == NULL;
    // 客户端没有执行事务
    if (!(c->flags & CLIENT_MULTI)) {
        addReplyError(c, "EXEC without MULTI");
        return;
    }

    /*
     * 检查是否需要阻止事务执行，因为：
     *
     * 1) Some WATCHed key was touched.
     *    有被监视的键已经被修改了
     *
     * 2) There was a previous error while queueing commands.
     *    命令在入队时发生错误
     *    （注意这个行为是 2.6.4 以后才修改的，之前是静默处理入队出错命令）
     * 第一种情况返回多个批量回复的空对象
     * 而第二种情况则返回一个 EXECABORT 错误
     */
    if (c->flags & (CLIENT_DIRTY_CAS | CLIENT_DIRTY_EXEC)) {
        addReply(c, c->flags & CLIENT_DIRTY_EXEC ? shared.execaborterr :
                    shared.nullarray[c->resp]);
        // 取消事务
        discardTransaction(c);
        goto handle_monitor;
    }

    uint64_t old_flags = c->flags;

    /* we do not want to allow blocking commands inside multi */
    c->flags |= CLIENT_DENY_BLOCKING;

    // 已经可以保证安全性了，取消客户端对所有键的监视
    unwatchAllKeys(c); /* Unwatch ASAP otherwise we'll waste CPU cycles */

    server.in_exec = 1;
    /*因为事务中的命令在执行时可能会修改命令和命令的参数，所以为了正确地传播命令，需要现备份这些命令和参数。*/
    orig_argv = c->argv;
    orig_argc = c->argc;
    orig_cmd = c->cmd;
    addReplyArrayLen(c, c->mstate.count);
    // 执行事务中的命令
    for (j = 0; j < c->mstate.count; j++) {
        /*因为 Redis 的命令必须在客户端的上下文中执行，
         * 所以要将事务队列中的命令、命令参数等设置给客户端*/
        c->argc = c->mstate.commands[j].argc;
        c->argv = c->mstate.commands[j].argv;
        c->cmd = c->mstate.commands[j].cmd;

        /* ACL permissions are also checked at the time of execution in case
         * they were changed after the commands were ququed. */
        int acl_errpos;
        int acl_retval = ACLCheckCommandPerm(c, &acl_errpos);
        if (acl_retval == ACL_OK && c->cmd->proc == publishCommand)
            acl_retval = ACLCheckPubsubPerm(c, 1, 1, 0, &acl_errpos);
        if (acl_retval != ACL_OK) {
            char *reason;
            switch (acl_retval) {
                case ACL_DENIED_CMD:
                    reason = "no permission to execute the command or subcommand";
                    break;
                case ACL_DENIED_KEY:
                    reason = "no permission to touch the specified keys";
                    break;
                case ACL_DENIED_CHANNEL:
                    reason = "no permission to publish to the specified channel";
                    break;
                default:
                    reason = "no permission";
                    break;
            }
            addACLLogEntry(c, acl_retval, acl_errpos, NULL);
            addReplyErrorFormat(c,
                                "-NOPERM ACLs rules changed between the moment the "
                                "transaction was accumulated and the EXEC call. "
                                "This command is no longer allowed for the "
                                "following reason: %s", reason);
        } else {
            call(c, server.loading ? CMD_CALL_NONE : CMD_CALL_FULL);
            serverAssert((c->flags & CLIENT_BLOCKED) == 0);
        }

        /* 因为执行后命令、命令参数可能会被改变
         * 所以这里需要更新事务队列中的命令和参数
         * 确保附属节点和 AOF 的数据一致性*/
        c->mstate.commands[j].argc = c->argc;
        c->mstate.commands[j].argv = c->argv;
        c->mstate.commands[j].cmd = c->cmd;
    }

    // restore old DENY_BLOCKING value
    if (!(old_flags & CLIENT_DENY_BLOCKING))
        c->flags &= ~CLIENT_DENY_BLOCKING;
    // 还原命令、命令参数
    c->argv = orig_argv;
    c->argc = orig_argc;
    c->cmd = orig_cmd;
    // 清理事务状态
    discardTransaction(c);

    /* 事务执行的命令要往其他节点传播，确保服务器和 AOF 文件以及附属节点的数据一致性。 */
    if (server.propagate_in_transaction) {
        int is_master = server.masterhost == NULL;
        // 将服务器设为脏，确保 EXEC 命令也会被传播
        server.dirty++;
        //这里会走下面方法的else逻辑，传播 exec 命令
        beforePropagateMultiOrExec(0);
        /* 如果在MULTI/EXEC块中，该实例突然从主实例切换到从实例(使用SLAVEOF命令)，则初始MULTI会传播到复制积压中，
         * 但其余的不会。我们需要确保至少在最终执行时终止积压。 */
        if (server.repl_backlog && was_master && !is_master) {
            char *execcmd = "*1\r\n$4\r\nEXEC\r\n";
            feedReplicationBacklog(execcmd, strlen(execcmd));
        }
    }
    //修改事务提交状态
    server.in_exec = 0;

    handle_monitor:
    /* Send EXEC to clients waiting data from MONITOR. We do it here
     * since the natural order of commands execution is actually:
     * MUTLI, EXEC, ... commands inside transaction ...
     * Instead EXEC is flagged as CMD_SKIP_MONITOR in the command
     * table, and we do it here with correct ordering. */
    if (listLength(server.monitors) && !server.loading)
        replicationFeedMonitors(c, server.monitors, c->db->id, c->argv, c->argc);
}

/* ===================== WATCH (CAS alike for MULTI/EXEC) ===================
 *
 * The implementation uses a per-DB hash table mapping keys to list of clients
 * WATCHing those keys, so that given a key that is going to be modified
 * we can mark all the associated clients as dirty.
 *
 * Also every client contains a list of WATCHed keys so that's possible to
 * un-watch such keys when the client is freed or when UNWATCH is called. */

/* In the client->watched_keys list we need to use watchedKey structures
 * as in order to identify a key in Redis we need both the key name and the
 * DB */
/*每个数据库redisDb都保存着一个watched_keys字典，
 * 字典的键是被WATCH命令所监视的某个键，值是一个链表，
 * 该链表记录着所有监视该键的客户端。
 * 客户端状态下也存放着被该客户端监视的键的链表集合。
 */
typedef struct watchedKey {
    // 被监视的键
    robj *key;
    // 键所在的数据库
    redisDb *db;
} watchedKey;

/* 让客户端监视给定的键 key */
void watchForKey(client *c, robj *key) {
    list *clients = NULL;
    listIter li;
    listNode *ln;
    watchedKey *wk;

    /* 检查 key 是否已经保存在 watched_keys 链表中，如果是的话，直接返回。 */
    listRewind(c->watched_keys, &li);
    while ((ln = listNext(&li))) {
        wk = listNodeValue(ln);
        if (wk->db == c->db && equalStringObjects(key, wk->key))
            return; /* Key already watched */
    }
    /* 检查 key 是否存在于数据库的 watched_keys 字典中 */
    clients = dictFetchValue(c->db->watched_keys, key);
    // 如果不存在的话，添加它
    if (!clients) {
        // 值为链表
        clients = listCreate();
        // 关联键值对到字典
        dictAdd(c->db->watched_keys, key, clients);
        incrRefCount(key);
    }
    // 将客户端添加到链表的末尾
    listAddNodeTail(clients, c);
    /* 将新 watchedKey 结构添加到客户端 watched_keys 链表的表尾 */
    wk = zmalloc(sizeof(*wk));
    wk->key = key;
    wk->db = c->db;
    incrRefCount(key);
    listAddNodeTail(c->watched_keys, wk);
}

/* 取消客户端对所有键的监视。 清除客户端事务状态的任务由调用者执行。 */
void unwatchAllKeys(client *c) {
    listIter li;
    listNode *ln;
    // 没有键被监视，直接返回
    if (listLength(c->watched_keys) == 0) return;
    // 遍历链表中所有被客户端监视的键
    listRewind(c->watched_keys, &li);
    while ((ln = listNext(&li))) {
        list *clients;
        watchedKey *wk;
        /* 从数据库的 watched_keys 字典的 key 键中删除链表里包含的客户端节点 */
        wk = listNodeValue(ln);
        // 取出客户端链表
        clients = dictFetchValue(wk->db->watched_keys, wk->key);
        serverAssertWithInfo(c, NULL, clients != NULL);
        // 删除链表中的客户端节点
        listDelNode(clients, listSearchKey(clients, c));
        /* 如果链表已经被清空，那么删除这个键*/
        if (listLength(clients) == 0) dictDelete(wk->db->watched_keys, wk->key);
        /* 从链表中移除 key 节点 */
        listDelNode(c->watched_keys, ln);
        decrRefCount(wk->key);
        zfree(wk);
    }
}

/* 触发监视机制 */
void touchWatchedKey(redisDb *db, robj *key) {
    list *clients;
    listIter li;
    listNode *ln;
    // 字典为空，没有任何键被监视
    if (dictSize(db->watched_keys) == 0) return;
    // 获取所有监视这个键的客户端
    clients = dictFetchValue(db->watched_keys, key);
    if (!clients) return;

    /*遍历所有客户端，打开他们的 REDIS_DIRTY_CAS 标识*/
    listRewind(clients, &li);
    while ((ln = listNext(&li))) {
        client *c = listNodeValue(ln);

        c->flags |= CLIENT_DIRTY_CAS;
    }
}

/* 当一个数据库被 FLUSHDB 或者 FLUSHALL 清空时，它数据库内的所有 key 都应该被触碰。
 * dbid 参数指定要被 FLUSH 的数据库。
 * 如果 dbid 为 -1 ，那么表示执行的是 FLUSHALL ，所有数据库都将被 FLUSH*/
void touchWatchedKeysOnFlush(int dbid) {
    listIter li1, li2;
    listNode *ln;
    //遍历所有客户端，然后遍历客户端监视的键，再让相应的客户端变为 DIRTY
    /* 遍历所有客户端*/
    listRewind(server.clients, &li1);
    while ((ln = listNext(&li1))) {
        client *c = listNodeValue(ln);
        // 遍历客户端监视的键
        listRewind(c->watched_keys, &li2);
        while ((ln = listNext(&li2))) {
            // 取出监视的键和键的数据库
            watchedKey *wk = listNodeValue(ln);
            // 如果数据库号码相同，或者执行的命令为 FLUSHALL，那么将客户端设置为 REDIS_DIRTY_CAS
            if (dbid == -1 || wk->db->id == dbid) {
                if (dictFind(wk->db->dict, wk->key->ptr) != NULL)
                    c->flags |= CLIENT_DIRTY_CAS;
            }
        }
    }
}

void watchCommand(client *c) {
    int j;
    // 不能在事务开始后执行
    if (c->flags & CLIENT_MULTI) {
        addReplyError(c, "WATCH inside MULTI is not allowed");
        return;
    }
    // 监视输入的任意个键
    for (j = 1; j < c->argc; j++)
        watchForKey(c, c->argv[j]);
    addReply(c, shared.ok);
}

void unwatchCommand(client *c) {
    // 取消客户端对所有键的监视
    unwatchAllKeys(c);
    c->flags &= (~CLIENT_DIRTY_CAS);
    // 重置状态
    addReply(c, shared.ok);
}
